package com.zan.rocketmqdemo.demo;

import com.zan.rocketmqdemo.constant.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.List;

/**
 * @Author Zan
 * @Create 2024/8/28 15:56
 * @ClassName: SimpleTest
 * @Description : 同步消息
 */
public class SimpleTest {

    /**
     * 发送消息
     */
    @Test
    public void simpleProducer() throws Exception {
        // 创建一个生产者（指定一个组名）
        DefaultMQProducer producer = new DefaultMQProducer("test-producer-group", true);
        // 连接 namesrv
        producer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        // 启动
        producer.start();
        // 创建一个消息
        for (int i = 0; i < 10; i++) {
            Message message = new Message("testTopic", "我是一个简单的消息".getBytes());
            // 发送消息
            SendResult sendResult = producer.send(message);
            System.out.println(sendResult.getSendStatus());
        }
        // 关闭生产者
        producer.shutdown();
    }

    /**
     * 消费消息
     */
    @Test
    public void simpleConsumer() throws Exception {
        // 创建一个消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");
        // push 就是 pull，基于pull模式的
//        DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer("test-consumer-group");
        // 连接 namesrv
        consumer.setNamesrvAddr(MqConstant.NAME_SRV_ADDR);
        // 订阅消息 - 订阅主题， * 表示订阅这个主题中的所有的消息，后期会有消息过滤
        consumer.subscribe("testTopic", "*");
        // 设置一个消息监听器（一直监听，异步回调方式）
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 这个就是消费的方法（一般是业务处理)
                System.out.println("我是消费者");
                System.out.println("消息内容：" + new String(msgs.get(0).getBody()));
                System.out.println("消费上下文：" + context);
                // 返回值非常关键 CONSUME_SUCCESS：成功消费（消息会从MQ队列中出队），RECONSUME_LATER：消费失败，稍后重新尝试消费（消息会重新回到MQ队列，供下次重新消费用，给其他消费者，或者自己进行消费）
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
        // 挂起当前的JVM
        System.in.read();
    }
}
